package com.cqrcb.cloud.config;

import java.time.Duration;
import java.util.Collections;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer.StreamMessageListenerContainerOptions;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.ErrorHandler;

@Component
public class StreamConsumerRunner implements ApplicationRunner, DisposableBean {

	static final Logger LOGGER = LoggerFactory.getLogger(StreamConsumerRunner.class);


	@Autowired
	private ThreadPoolTaskExecutor threadPoolTaskExecutor;

	@Autowired
	private RedisConnectionFactory redisConnectionFactory;

	@Autowired
	private ListenerMessage streamMessageListener;

	@Autowired
	private StringRedisTemplate stringRedisTemplate;

	private StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer;

	@Override
	public void run(ApplicationArguments args) throws Exception {

		String channelName = "channel";
		String groupName = "group";
		String consumerName = "consumer-1";

		// 初始化stream
		if(!this.stringRedisTemplate.hasKey(channelName)) {
			RecordId recordId = this.stringRedisTemplate.opsForStream().add(channelName, Collections.singletonMap("tryCreat", ""));
			LOGGER.info("stream {} 初始化：{}", "channel", recordId);
		}

		// 初始化消费组
		try {
			String result = this.stringRedisTemplate.opsForStream().createGroup(channelName, "group");
			LOGGER.info("消费组 {} 创建：{}", "group", result);
		} catch (Exception e) {
			LOGGER.error("消费组创建失败:{}", e.getMessage());
		}

		StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> streamMessageListenerContainerOptions = StreamMessageListenerContainerOptions
				.builder()
				.batchSize(2)
				.executor(threadPoolTaskExecutor)
				.errorHandler(new ErrorHandler() {
					@Override
					public void handleError(Throwable t) {
						t.printStackTrace();
					}
				})
				.pollTimeout(Duration.ZERO)
				.serializer(new StringRedisSerializer())
				.build();

		// 创建 Container
		StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer = StreamMessageListenerContainer
				.create(this.redisConnectionFactory, streamMessageListenerContainerOptions);

		// 群组消费：开始接收消息，设置为手动消费
		streamMessageListenerContainer.receive(Consumer.from(groupName, consumerName),
				StreamOffset.create(channelName, ReadOffset.lastConsumed()), this.streamMessageListener);


		this.streamMessageListenerContainer = streamMessageListenerContainer;

		this.streamMessageListenerContainer.start();

	}

	@Override
	public void destroy() throws Exception {
		this.streamMessageListenerContainer.stop();
	}
}